package com.ruirui.demo;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Signal;
import reactor.core.publisher.SignalType;

import java.time.Duration;
import java.util.function.Consumer;

public class FluxDoxxx {
  /**
   * doOn开头 钩子函数 xxx操作完成后调用 在流上的感知都是发布者操作
   * 所有操作都是对上个流的感知
   */

    public static void main(String[] args) throws InterruptedException {
        // 流式操作都是以上一步流的接力
        Flux<Integer> flux = Flux.range(1,3);
        //链式api 下面的操作操作的的是上面的流 最终成为新流
        flux.delayElements(Duration.ofMillis(1000))
                //doOn开头 钩子函数 xxx操作完成后调用 所有操作都是对上个流的感知
                .doOnNext(e-> System.out.println("next"+e))
                .doOnEach(e->System.out.println("each"+e))
                //上流有错误就触发
                .doOnError(e->System.out.println("error"))
                //流的下一个元素到达的时候
                .subscribe(System.out::println);


        Thread.sleep(5000);

//        flux.delayElements(Duration.ofMillis(1000)).
//                doOnCancel(()->System.out.println("cancel"));
//        Thread.sleep(2000);
    }
}

